package task4

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
case class WaterMark(id:Int,time:Long)

/**
 * @author jhhe66
 * @date 2021/6/18 11:37
 */
object WaterMark {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 指定事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 水印生产周期
    env.getConfig.setAutoWatermarkInterval(200)
    // 设置并行度
    env.setParallelism(1)

    val dataDS: DataStream[String] = env.socketTextStream("centos001", 9999)
    // 将数据流封装成样例类
    val wcDS: DataStream[WaterMark] = dataDS.map { line =>
      val arr: Array[String] = line.split(",")
      WaterMark(arr(0).trim.toInt, arr(1).trim.toLong)
    }

    // 注册水印，设置允许吃到为2秒
    val waterMarkDS: DataStream[WaterMark] = wcDS.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[WaterMark](Time.seconds(2)) {
      // 返回事件时间字段
      override def extractTimestamp(t: WaterMark): Long = {
        t.time
      }
    })
    // 设置窗口为5秒
    val windowDS: WindowedStream[WaterMark, Tuple, TimeWindow] = waterMarkDS.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)))
    // apply方法，输出样例类对象
    val res = windowDS.apply(new WindowFunction[WaterMark, WaterMark, Tuple, TimeWindow] {
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[WaterMark], out: Collector[WaterMark]): Unit = {
        // 打印开始时间和结束时间
        println(s"windows开始时间${window.getStart}.window结束时间${window.getEnd}")
        // 输出样例类
        input.foreach(x => out.collect(x))
      }
    })

    res.print()
    env.execute()
  }
}


